package global

import (
	"fmt"
	redisBloom "github.com/RedisBloom/redisbloom-go"
	"github.com/go-redis/redis"
	rdsMod "github.com/gomodule/redigo/redis"
	cfg "github.com/ufoscout/go-up"
	"strings"
	"time"

	"github.com/spf13/cast"
)

var rds *redis.Client
var rdsCluster *redis.ClusterClient
var bloom *redisBloom.Client
var password string
var confFilePath string

func InitRedis(sFilePath string) (redis.UniversalClient, error) {
	fName := "InitRedis"
	isCluster := Config.GetString("redis.condition.cluster")
	SetCfgPath(sFilePath)
	if len(isCluster) <= 0 || !cast.ToBool(isCluster) {
		redis := doInit(
			Config.GetString("redis.host"),
			Config.GetString("redis.port"),
			Config.GetString("redis.passwd"),
			Config.GetInt("redis.db"))
		ping := redis.Ping().String()
		if ping != "ping: PONG" {
			return nil, fmt.Errorf("%s fail %v", fName, ping)
		}
		return redis, nil
	} else {
		fName := "InitRedis"
		redis := InitRedisCluster(
			Config.GetStringSlice("redisCluster.host", ","),
			Config.GetString("redis.passwd"))
		ping := redis.Ping().String()
		if ping != "ping: PONG" {
			return nil, fmt.Errorf("%s fail %v", fName, ping)
		}
		return redis, nil
	}
}

func doInit(Host, Port, Password string, DBindex int) *redis.Client {
	rds = redis.NewClient(&redis.Options{
		Addr:     Host + ":" + Port,
		Password: Password,
		DB:       DBindex,
	})
	return rds
}

func InitBloom() error {
	fName := "InitBloom"
	isCluster := Config.GetString("redis.condition.cluster")
	if len(isCluster) <= 0 || !cast.ToBool(isCluster) {
		bloom := InitBloomFilter(
			Config.GetString("redis.host"),
			Config.GetString("redis.port"),
			Config.GetString("redis.passwd"),
			Config.GetInt("redis.db"))
		if bloom == nil {
			return fmt.Errorf("%s fail", fName)
		}
		return nil
	} else {
		addrList := Config.GetStringSlice("redisCluster.host", ",")
		bloom := InitBloomFilterCluster(
			addrList[0],
			Config.GetString("redis.passwd"),
		)
		if bloom == nil {
			return fmt.Errorf("%s fail", fName)
		}
		return nil
	}
}

func InitBloomCluster() error {
	fName := "InitBloom"
	isCluster := Config.GetString("redis.condition.cluster")
	if len(isCluster) <= 0 || !cast.ToBool(isCluster) {
		bloom := InitBloomFilter(
			Config.GetString("redis.host"),
			Config.GetString("redis.port"),
			Config.GetString("redis.passwd"),
			Config.GetInt("redis.db"))
		if bloom == nil {
			return fmt.Errorf("%s fail", fName)
		}
		return nil
	} else {
		addrList := Config.GetStringSlice("redisCluster.host", ",")
		bloom := InitBloomFilterCluster(
			addrList[0],
			Config.GetString("redis.passwd"),
		)
		if bloom == nil {
			return fmt.Errorf("%s fail", fName)
		}
		return nil
	}
}

func InitBloomFilter(Host, Port, Password string, DBindex int) *redisBloom.Client {
	host := Host + ":" + Port
	p := &rdsMod.Pool{Dial: func() (rdsMod.Conn, error) {
		return rdsMod.Dial("tcp", host, rdsMod.DialPassword(Password), rdsMod.DialDatabase(DBindex))
	}}
	bloom = redisBloom.NewClientFromPool(p, "bloom-client")
	return bloom
}

func InitBloomFilterCluster(addr, Password string) *redisBloom.Client {
	p := &rdsMod.Pool{Dial: func() (rdsMod.Conn, error) {
		return rdsMod.Dial("tcp", addr, rdsMod.DialPassword(Password), rdsMod.DialDatabase(0))
	}}
	bloom = redisBloom.NewClientFromPool(p, "bloom-client")
	return bloom
}

func GetBloom() *redisBloom.Client {
	if bloom == nil {
		initConf()
		isCluster := Config.GetString("redis.condition.cluster")
		if len(isCluster) <= 0 || !cast.ToBool(isCluster) {
			bloom = InitBloomFilter(
				Config.GetString("redis.host"),
				Config.GetString("redis.port"),
				Config.GetString("redis.passwd"),
				Config.GetInt("redis.db"))

			return bloom
		} else {
			addrList := Config.GetStringSlice("redisCluster.host", ",")
			bloom = InitBloomFilterCluster(
				addrList[0],
				Config.GetString("redis.passwd"),
			)
			return bloom
		}
	}
	return bloom
}

func initConf() error {
	fName := "InitConf"
	config, err := cfg.NewGoUp().AddFile(confFilePath, false).Build()
	if err != nil {
		return fmt.Errorf("%s fail %s , file path is ； %s", fName, err, confFilePath)
	}
	Config = config
	return nil
}

func BloomExists(key, item string) (bool, error) {
	b, err := GetBloom().Exists(key, item)
	if err != nil {
		if strings.Contains(err.Error(), "MOVED") {
			changeBloomHostAfterError(err.Error()) //重连一次
		} else {
			fmt.Printf("bloom error:%s\n", err.Error())
		}
		return true, nil //出错当成存在，可进下一步，不然整个卡住
	}
	return b, err
}

func GetRds() redis.UniversalClient {
	if rds == nil {
		return rdsCluster
	}
	return rds
}

/**/
func BloomAdd(key, item string) (bool, error) {
	b, err := GetBloom().Add(key, item)
	if err != nil {
		if strings.Contains(err.Error(), "MOVED") {
			changeBloomHostAfterError(err.Error()) //重连一次
			return GetBloom().Add(key, item)
		} else {
			fmt.Printf("bloom error:%s\n", err.Error())
		}
		return true, nil
	}
	return b, err
}
func BloomBfAddMulti(key string, items []string) ([]int64, error) {
	b, err := GetBloom().BfAddMulti(key, items)
	if err != nil {
		if strings.Contains(err.Error(), "MOVED") {
			changeBloomHostAfterError(err.Error()) //重连一次
			return GetBloom().BfAddMulti(key, items)
		} else {
			fmt.Printf("bloom error:%s\n", err.Error())
			return []int64{}, nil
		}
	}
	return b, err
}
func changeBloomHostAfterError(errInfo string) {
	aString := strings.Split(errInfo, " ")
	sHost := aString[len(aString)-1]
	bloom = InitBloomFilterCluster(sHost, password)
	if bloom == nil {
		fmt.Printf("Error :bloom init error%s@%s\n", sHost, password)
	}
}

/*
 */
func InitRedisCluster(Host []string, Password string) *redis.ClusterClient {
	rdsCluster = redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:        Host,
		Password:     Password, // 设置密码
		PoolSize:     800,
		MinIdleConns: 300,
		DialTimeout:  3000 * time.Microsecond, // 设置连接超时
		ReadTimeout:  2000 * time.Microsecond, // 设置读取超时
		WriteTimeout: 2000 * time.Microsecond, // 设置写入超时
	})
	Log.Debugf("InitRedisCluster addrs:%s", Host)
	return rdsCluster
}

const (
	BLOOM_MOBILE     = "mobile"
	BLOOM_ID         = "id"
	BLOOM_ANCHOR     = "anchor_bloom"
	BLOOM_IPBLACK    = "ip_black"
	BLOOM_SEMCHANNEL = "sem_channel"
)

func SetCfgPath(f string) {
	confFilePath = f
}
